在 Day 25 與 Day 26 中，我們打造了完整的 Agent Trace 追蹤鏈與 Streamlit 戰情檢視器。然而，當無人值守管線真正運行於 GitHub Actions 或定時排程時，分佈式外部相依（External Dependencies）的脆弱性暴露無遺：

• Google AI Studio 免費額度限制：突發並發處理時，連續呼叫迅速撞上 15 RPM / TPM 速率天花板，收到 429 RESOURCE_EXHAUSTED。
• 雲端尖峰負載抖動：LLM 服務商在高峰期隨機拋出 503 Service Unavailable 或超時掛起。
• 管線骨牌效應：單一情報的 Reviewer 節點失敗，如果沒有退避重試，就會一路導致後續代理人無法執行，整批情報被粗暴地全部標記為 ERROR。

今天我們將在基礎代理人層（BaseAgent）與核心編排管線中，實作生產級的指數退避重試（Exponential Backoff）、抖動因子（Jitter）與模型動態降級策略（Model Fallback），讓情報雷達在惡劣的外部 API 環境下依然具備自癒彈性！

一、防禦架構設計：多層容錯與降級路徑

當代理人請求 LLM API 遭遇失敗時，系統不應立即放棄，而應依循以下三級防禦機制：

[代理人發起推論請求 (Primary Model: gemini-2.5-pro / 3.8-flash)]
                     │
                     ▼
             ┌───────────────┐
             │ 呼叫成功？    ├──────(是)─────► 正常進入下一節點
             └───────┬───────┘
                     │ (否: 429 / 503 / 網路超時)
                     ▼
      [第一道防線：指數退避重試 (Exponential Backoff)]
      - 重試 3 次，間隔時間指數遞增 (3s -> 6s -> 12s)
      - 加入隨機抖動 Jitter，避免群發共振
                     │
                     ▼ 仍持續失敗？
      [第二道防線：模型動態降級 (Model Fallback)]
      - 自動降級至高吞吐、低負載模型 (例如切換為 gemini-2.5-flash-lite)
      - 重新嘗試推論
                     │
                     ▼ 依舊崩潰？
      [第三道防線：保險絲熔斷與狀態保護]
      - 記錄失敗 TraceSpan (含錯誤代碼與重試次數)
      - 狀態安全回退，標記 ERROR 並觸發 Discord 靜默告警

二、基礎工具實作：tenacity 指數退避裝飾器

在 Python 中，處理重試邏輯的最佳實踐是使用經過廣泛驗證的 tenacity 函式庫。

在 pyproject.toml 與 requirements.txt 中加入依賴：

tenacity>=8.3.0

指數退避的核心公式：

$$\text{Wait Time} = \min(\text{max_wait}, \text{multiplier} \times 2^{\text{attempt}}) \pm \text{jitter}$$

• Multiplier（乘數）：基準倍率（例如設為 2）。
• Min / Max Wait：重試最短等待（3 秒）與最長上限（15 秒），避免無休止等待導致 GitHub Actions 超時。
• Random Jitter（隨機抖動）：避免同一批多個 Worker 同時重試，再次撞爛 API。

三、適配器層升級：在 BaseAgent 注入重試與動態降級

我們修改代理人的基礎類別 src/agents/base.py，使所有子代理人（Reviewer、DomainExpert、Synthesizer、Editor）天然繼承重試與自動降級的能力：

# src/agents/base.py
from __future__ import annotations

import logging
from typing import Any
from google import genai
from google.genai.errors import APIError
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log,
)

logger = logging.getLogger("agent.base")

# 定義觸發重試的特定例外（排除使用者的 Prompt 語法錯誤）
RETRYABLE_ERRORS = (APIError, TimeoutError, ConnectionError)

class BaseAgent:
    def __init__(
        self,
        client: genai.Client,
        primary_model: str = "gemini-2.5-flash",
        fallback_model: str = "gemini-2.5-flash-lite",
        max_retries: int = 3
    ):
        self.client = client
        self.primary_model = primary_model
        self.fallback_model = fallback_model
        self.max_retries = max_retries

    @retry(
        retry=retry_if_exception_type(RETRYABLE_ERRORS),
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=2, min=3, max=15),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True
    )
    def _call_model_with_retry(self, model_name: str, contents: Any, config: Any) -> Any:
        """具備指數退避重試的底層調用"""
        return self.client.models.generate_content(
            model=model_name,
            contents=contents,
            config=config
        )

    def generate_with_fallback(self, contents: Any, config: Any) -> tuple[Any, str]:
        """
        雙軌執行：優先使用主模型，若重試耗盡仍遭拒絕，則自動降級至備用模型。
        回傳：(產生結果, 最終使用的模型名稱)
        """
        try:
            # 1. 嘗試調用主模型 (含 3 次指數退避)
            response = self._call_model_with_retry(self.primary_model, contents, config)
            return response, self.primary_model
        except RETRYABLE_ERRORS as primary_err:
            logger.warning(
                f"主模型 [{self.primary_model}] 重試耗盡失敗: {primary_err}。"
                f"正在觸發自動降級至備用模型 [{self.fallback_model}]..."
            )

            try:
                # 2. 自動降級調用輕量備用模型 (同樣具備防禦重試)
                response = self._call_model_with_retry(self.fallback_model, contents, config)
                logger.info(f"備用模型 [{self.fallback_model}] 呼叫成功！成功拯救本次推論任務。")
                return response, self.fallback_model
            except Exception as fallback_err:
                logger.critical(f"備用模型 [{self.fallback_model}] 亦告崩潰: {fallback_err}")
                raise fallback_err

四、代理人子類別改造：以 ReviewerAgent 為例

有了 BaseAgent 的強大基底，子代理人的寫法變得極其乾淨：

# src/agents/reviewer.py
from src.agents.base import BaseAgent
from src.core.models import IntelligenceItem

class ReviewerAgent(BaseAgent):
    def __init__(self, client, model_name: str = "gemini-2.5-flash", threshold: int = 4):
        # 設定主模型與降級模型
        super().__init__(
            client=client,
            primary_model=model_name,
            fallback_model="gemini-2.5-flash-lite",
            max_retries=3
        )
        self.threshold = threshold

    def evaluate(self, item: IntelligenceItem) -> tuple[IntelligenceItem, bool]:
        prompt = f"請審查此技術情報價值...\n標題: {item.source_title}\n內容: {item.raw_content}"
        
        # 呼叫父類的動態降級推論
        response, used_model = self.generate_with_fallback(
            contents=prompt,
            config={"response_mime_type": "application/json"}
        )

        # 紀錄實際使用的模型於追蹤中
        item.metadata["reviewer_used_model"] = used_model
        
        # 解析分數與決策邏輯...
        ...
        return item, True

五、編排管線防禦：pipeline.py 的原子級保險絲

在單一項目處理層面，我們在 process_item 外圍建構完整的保險絲機制，防止偶發錯誤破壞試算表狀態：

# src/services/pipeline.py (節錄)
def process_item(self, item: IntelligenceItem) -> IntelligenceItem:
    logger.info(f"開始處理情資 [{item.entry_id}]: {item.source_title}")
    
    try:
        # Step 1: Reviewer (初審熔斷)
        item, should_continue = self.reviewer.evaluate(item)
        if not should_continue:
            logger.info(f"[{item.entry_id}] 未達門檻，標記 ARCHIVED")
            item.status = ItemStatus.ARCHIVED
            return item

        # Step 2: Domain Expert (量化指標萃取)
        item, _ = self.domain_expert.extract_metrics(item)

        # Step 3: Synthesizer (二階跨界推演)
        item, _ = self.synthesizer.synthesize(item)

        # Step 4: Editor (摘要與標籤精煉)
        item, _ = self.editor.edit(item)

        # 全部成功，標記為 PROCESSED
        item.status = ItemStatus.PROCESSED
        return item

    except Exception as exc:
        logger.error(f"處理情資 [{item.entry_id}] 遭遇未捕獲例外: {exc}", exc_info=True)
        # 標記為 ERROR，避免永久卡在 ANALYZING
        item.status = ItemStatus.ERROR
        item.metadata["error_reason"] = str(exc)
        return item

六、實戰成效驗證

在本地模擬 503 服務不可用情境下執行管線：

2026-10-12 00:05:01 - [WARNING] - 主模型 [gemini-2.5-flash] 呼叫失敗: 503 Service Unavailable. 正在進行第 1 次重試 (等待 3.2 秒)...
2026-10-12 00:05:04 - [WARNING] - 主模型 [gemini-2.5-flash] 呼叫失敗: 503 Service Unavailable. 正在進行第 2 次重試 (等待 6.8 秒)...
2026-10-12 00:05:11 - [WARNING] - 主模型 [gemini-2.5-flash] 重試耗盡失敗。正在觸發自動降級至備用模型 [gemini-2.5-flash-lite]...
2026-10-12 00:05:12 - [INFO] - 備用模型 [gemini-2.5-flash-lite] 呼叫成功！成功拯救本次推論任務。
2026-10-12 00:05:13 - [INFO] - [165ebc41] 成功評審完成 | 分數: 5 | 狀態: PROCESSED

原本會直接翻車標記為 ERROR 的情資，在退避重試與降級防禦的保護下，順利完成研報提煉並轉化為 PROCESSED！

總結與下一步

今天我們為情報雷達補齊了生產環境最關鍵的韌性拼圖：

• 釐清了 消費者訂閱產品與開發者 API 配額的本質區別。
• 在 BaseAgent 整合 tenacity，實現了 指數退避 + 隨機抖動 + 模型自動降級。
• 保證了無人值守巡航在外部 API 偶發抖動時的生存率與資料完整性。

目前系統架構已經非常龐大，涵蓋了 4 種轉接器（Sheets, Discord, Obsidian, Notion）與多個核心服務。如果每次微調代碼都要跑真正的外部 API 測試，不僅耗時且浪費配額。

在 Day 28，我們將進入專業工程品質把關：「防禦型合約測試：針對 Obsidian / Notion 適配器的自動化單元測試」，使用 Mock 技術確保所有外掛端口的契約完整性！